﻿/*
* Copyright: JessMA Open Source (ldcsaa@gmail.com)
*
* Author	: Bruce Liang
* Website	: http://www.jessma.org
* Project	: https://github.com/ldcsaa
* Blog		: http://www.cnblogs.com/ldcsaa
* Wiki		: http://www.oschina.net/p/hp-socket
* QQ Group	: 75375912, 44636872
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
*      http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

#include "IODispatcher.h"
#include "FuncHelper.h"

#include <signal.h>
#include <pthread.h>

/**
 * pHandler: IO事件调度相应的实现
 * iWorkerMaxEvents: 处理线程中单次处理事件最大限制
 * iWorkers: 工作线程数量
 * llTimerInterval: 定时器间隔，windows平台上不支持
 */
BOOL CIODispatcher::Start(IIOHandler* pHandler, int iWorkerMaxEvents, int iWorkers, LLONG llTimerInterval)
{
	ASSERT_CHECK_EINVAL(pHandler && iWorkerMaxEvents >= 0 && iWorkers >= 0);
	CHECK_ERROR(!HasStarted(), ERROR_INVALID_STATE);

	//设置默认值
	if(iWorkerMaxEvents == 0)	iWorkerMaxEvents = DEF_WORKER_MAX_EVENTS;
	if(iWorkers == 0)			iWorkers = DEFAULT_WORKER_THREAD_COUNT;

	m_iMaxEvents = iWorkerMaxEvents;
	m_iWorkers	 = iWorkers;
	m_pHandler	 = pHandler;

	//创建一个epoll监听机制的描述符
	//EPOLL_CLOEXEC: fork子进程后在子进程中直接执行close关掉无用的文件描述符，然后再执行exec
	m_epoll = epoll_create1(EPOLL_CLOEXEC);
	CHECK_ERROR_FD(m_epoll);

	//主要用于一些调度命令的处理，TDispCommand
	//创建一个文件描述符用于事件通知，类似于pipe
	//EFD_NONBLOCK
	m_evCmd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);

	if(IS_INVALID_FD(m_evCmd))
		goto START_ERROR;

	//将m_evCmd添加至m_epoll监听中
	if(!VERIFY(AddFD(m_evCmd, EPOLLIN | EPOLLET, &m_evCmd)))
		goto START_ERROR;

	//EFD_SEMAPHORE：每次读取只会做信号量的递减操作；如：写入了一个10，读取一次变为9
	m_evExit = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC | EFD_SEMAPHORE);

	if(IS_INVALID_FD(m_evExit))
		goto START_ERROR;

	//将m_evExit添加至m_epoll监听中
	if(!VERIFY(AddFD(m_evExit, EPOLLIN, &m_evExit)))
		goto START_ERROR;

	//创建定时器，支持纳秒级
	if(llTimerInterval > 0)
	{
		m_evTimer = timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);;

		if(IS_INVALID_FD(m_evTimer))
			goto START_ERROR;

		itimerspec its;

		::MillisecondToTimespec(llTimerInterval, its.it_value);
		::MillisecondToTimespec(llTimerInterval, its.it_interval);

		if(!VERIFY_IS_NO_ERROR(timerfd_settime(m_evTimer, 0, &its, nullptr)))
			goto START_ERROR;
		//添加至m_epoll
		if(!VERIFY(AddFD(m_evTimer, EPOLLIN | EPOLLET, &m_evTimer)))
			goto START_ERROR;
	}

	sigset_t ss;
	sigemptyset(&ss);
	sigaddset(&ss, SIGPIPE); 
	//屏蔽SIGPIPE进程信号
	//主要是TCP通信中服务器尝试send到一个disconnected socket上，就会让底层抛出一个SIGPIPE信号，而默认行为与期望不符，所以直接屏蔽
	VERIFY_IS_NO_ERROR(pthread_sigmask(SIG_BLOCK, &ss, nullptr));

	//创建具体的IO处理线程
	m_pWorkers = make_unique<CWorkerThread[]>(m_iWorkers);

	//执行线程
	for(int i = 0; i < m_iWorkers; i++)
	{
		if(!VERIFY(m_pWorkers[i].Start(this, &CIODispatcher::WorkerProc)))
			goto START_ERROR;
	}
	//HP-Socket框架释放了调用此方法的线程
	return TRUE;

START_ERROR:
	EXECUTE_RESTORE_ERROR(Stop(FALSE));
	return FALSE;
}

/**
 * 停止HP-Socket框架工作
 * bCheck: 停止前是否检查已有状态, 主要防止重复调用
 */
BOOL CIODispatcher::Stop(BOOL bCheck)
{
	if(bCheck) CHECK_ERROR(HasStarted(), ERROR_INVALID_STATE);

	BOOL isOK = TRUE;

	//关闭工作线程
	if(m_pWorkers)
	{
		//向m_evExit发送一个信号量，要注意EFD_SEMAPHORE标识
		//当工作线程读取时只会做递减操作
		isOK &= IS_NO_ERROR(eventfd_write(m_evExit, m_iWorkers));

		//等待工作线程，这里会阻塞调用此方法的线程
		for(int i = 0; i < m_iWorkers; i++)
			isOK &= m_pWorkers[i].Join();
	}

	//释放调度命令队列
	if(!m_queue.IsEmpty())
	{
		TDispCommand* pCmd = nullptr;

		while(m_queue.PopFront(&pCmd))
			TDispCommand::Destruct(pCmd);

		VERIFY(m_queue.IsEmpty());
	}

	//关闭文件描述符
	//注意linux上的close关闭只会做引用计数递减操作，当为0时才会执行关闭
	if(IS_VALID_FD(m_evExit))
		isOK &= IS_NO_ERROR(close(m_evExit));

	if(IS_VALID_FD(m_evCmd))
		isOK &= IS_NO_ERROR(close(m_evCmd));

	if(IS_VALID_FD(m_evTimer))
		isOK &= IS_NO_ERROR(close(m_evTimer));

	if(IS_VALID_FD(m_epoll))
		isOK &= IS_NO_ERROR(close(m_epoll));

	Reset();

	return isOK;
}

VOID CIODispatcher::Reset()
{
	m_iWorkers	= 0;
	m_iMaxEvents= 0;
	m_pHandler	= nullptr;
	m_pWorkers	= nullptr;
	m_epoll		= INVALID_FD;
	m_evCmd		= INVALID_FD;
	m_evExit	= INVALID_FD;
	m_evTimer	= INVALID_FD;
}

/**
 * 发送调度命令，添加至一个CASQueue队列中
 * t: type 事件类型
 * wp: 附带参数1
 * lp: 附带参数2
 */
BOOL CIODispatcher::SendCommand(USHORT t, UINT_PTR wp, UINT_PTR lp)
{
	return SendCommand(TDispCommand::Construct(t, wp, lp));
}

/**
 * 添加一个调度命令至CASQueue队列中
 */
BOOL CIODispatcher::SendCommand(TDispCommand* pCmd)
{
	m_queue.PushBack(pCmd);
	return VERIFY_IS_NO_ERROR(eventfd_write(m_evCmd, 1));
}

/**
 * 对m_epoll执行op操作
 * fd： 相应的文件资源描述符
 * op： 执行的操作{EPOLL_CTL_ADD, EPOLL_CTL_MOD, EPOLL_CTL_DEL}
 * mask：事件掩码，如：EPOLLIN, EPOLLOUT等
 * pv：附带参数
 */
BOOL CIODispatcher::CtlFD(FD fd, int op, UINT mask, PVOID pv)
{
	epoll_event evt = {mask, pv};
	return IS_NO_ERROR(epoll_ctl(m_epoll, op, fd, &evt));
}

/**
 * HP-Socket工作线程具体的处理过程
 */
int CIODispatcher::WorkerProc(PVOID pv)
{
	BOOL bRun							= TRUE;
	//最大单位可处理的epoll事件
	unique_ptr<epoll_event[]> pEvents	= make_unique<epoll_event[]>(m_iMaxEvents);

	while(bRun)
	{
		//epoll_pwait是线程安全的
		//INFINITE: 无限期阻塞当前工作线程，若是调用Stop会使线程退出阻塞状态执行ProcessExit
		//rs：返回值表示当前线程接收到了多少事件
		//不会出现对于一个Socket描述符的多线程问题，同一时刻一个Socket只会被一个线程处理，主要是epoll机制的ONESHOT
		int rs = NO_EINTR_INT(epoll_pwait(m_epoll, pEvents.get(), m_iMaxEvents, INFINITE, nullptr));

		//当m_epoll不是一个epoll的文件资源描述符或m_iMaxEvents<=0时，rs将返回为0
		//当前线程errno被设置为某些错误时rs返回-1
		if(rs <= TIMEOUT)
			ERROR_ABORT();

		//触发具体的事件后的处理
		for(int i = 0; i < rs; i++)
		{
			//相应的事件集
			UINT events	= pEvents[i].events;
			//相应文件描述符附带的参数，m_evCmd，m_evExit，m_evTimer
			//不是它们则是Socket I/O相关
			PVOID ptr	= pEvents[i].data.ptr;

			if(ptr == &m_evCmd)
				ProcessCommand(events); //触发调度命令事件
			else if(ptr == &m_evTimer)
				ProcessTimer(events); //定时器出发
			else if(ptr == &m_evExit)
				bRun = ProcessExit(events); //触发线程退出事件重置bRun状态
			else
				ProcessIo(ptr, events);
		}
	}

	//线程退出触发回调，一般用与大循环的生命周期结束时的业务处理
	m_pHandler->OnDispatchThreadEnd(SELF_THREAD_ID);

	return 0;
}

/**
 * 调度命令的处理过程
 */
BOOL CIODispatcher::ProcessCommand(UINT events)
{
	//判断events是否正确
	if(events & _EPOLL_ALL_ERROR_EVENTS)
		ERROR_ABORT();

	//判断events不为EPOLLIN即：关联的fd不能read
	if(!(events & EPOLLIN))
		return FALSE;

	BOOL isOK = TRUE;

	eventfd_t v;

	//读取成功返回0
	int rs = eventfd_read(m_evCmd, &v);

	//判断是否为0即：是否成功
	if(IS_NO_ERROR(rs))
	{
		ASSERT(v > 0);

		TDispCommand* pCmd = nullptr;

		//从CASQueue队列中取出调度命令
		while(m_queue.PopFront(&pCmd))
		{
			//执行相应回调
			m_pHandler->OnCommand(pCmd);
			//销毁调度命令
			TDispCommand::Destruct(pCmd);
		}
	}
	else if(IS_HAS_ERROR(rs))
	{
		//内部出现问题
		ASSERT(IS_WOULDBLOCK_ERROR());

		isOK = FALSE;
	}

	return isOK;
}

/**
 * 定时器处理过程
 */
BOOL CIODispatcher::ProcessTimer(UINT events)
{
	//固定的长度，用于对定时器读取时的传参
	static const SSIZE_T SIZE = sizeof(LLONG);

	if(events & _EPOLL_ALL_ERROR_EVENTS)
		ERROR_ABORT();

	if(!(events & EPOLLIN))
		return TRUE;

	BOOL isOK = TRUE;
	//用于存储定时器过期时的时间
	LLONG llExpirations;
	
	if(read(m_evTimer, &llExpirations, SIZE) == SIZE)
		m_pHandler->OnTimer(llExpirations); //响应回调
	else
	{
		ASSERT(IS_WOULDBLOCK_ERROR());

		isOK = FALSE;
	}

	return isOK;
}

/**
 * 退出处理过程
 */
BOOL CIODispatcher::ProcessExit(UINT events)
{
	if(events & _EPOLL_ALL_ERROR_EVENTS)
		ERROR_ABORT();

	if(!(events & EPOLLIN))
		return TRUE;

	BOOL bRun = TRUE;

	eventfd_t v;

	//读m_evExit，因为具有EFD_SEMAPHORE，所以内部只会递减信号量
	int rs = eventfd_read(m_evExit, &v);

	//正确读取次数不会为true，即rs == -1
	if(IS_HAS_ERROR(rs))
		ASSERT(IS_WOULDBLOCK_ERROR());
	else
	{
		//正确的读取rs值为1
		ASSERT(v == 1);
		bRun = FALSE;
	}

	//返回结果影响HPSocket工作线程的生命周期
	return bRun;
}

/**
 * Socket IO处理过程
 * pv: 附带的指针参数，帮助后续判断
 * events: 事件集
 */
BOOL CIODispatcher::ProcessIo(PVOID pv, UINT events)
{
	//在处理相应Socket IO前，调用相应IO实现类的回调
	if(!m_pHandler->OnBeforeProcessIo(pv, events))
		return FALSE;

	//处理相应Socket IO
	BOOL rs = DoProcessIo(pv, events);
	//在处理相应Socket IO后，调用相应IO实现类的回调
	m_pHandler->OnAfterProcessIo(pv, events, rs);

	//返回结果不影响HPSocket工作线程的生命周期
	return rs;
}

BOOL CIODispatcher::DoProcessIo(PVOID pv, UINT events)
{
	//一般为内部错误，关闭相应的Socket
	if(events & EPOLLERR)
		return m_pHandler->OnError(pv, events);
	//是否为外带数据，并执行相应回调
	if((events & EPOLLPRI) && !m_pHandler->OnReadyPrivilege(pv, events))
		return FALSE;
	//接收数据事件，并执行相应回调
	if((events & EPOLLIN) && !m_pHandler->OnReadyRead(pv, events))
		return FALSE;
	//发送数据事件，并执行相应回调
	if((events & EPOLLOUT) && !m_pHandler->OnReadyWrite(pv, events))
		return FALSE;
	//发生了挂起，关闭相应的Socket
	if((events & (_EPOLL_HUNGUP_EVENTS)) && !m_pHandler->OnHungUp(pv, events))
		return FALSE;

	//返回结果不影响HPSocket工作线程的生命周期
	return TRUE;
}
